#include "AsyncComm.h"
#include "communication/DataPacket.h"
#include "communication/proto/opcode.h"
#include "Session.h"
#define TIME_TIMEOUT (30*1000)
#define PACK_PACKET 0


AsyncComm::AsyncComm() : 
m_fd(INVALID_FD), 
m_connected(false),
m_timeSendPing(0),
m_pSession(NULL)
{
	// Allocate Buffers
	readBuffer.Allocate(CLIENTSOCKET_RECVBUF_SIZE);
	writeBuffer.Allocate(CLIENTSOCKET_SENDBUF_SIZE);

	_Reset();
}

bool AsyncComm::InitFD(HANDLE_FD fd, Session* pSession)
{
	m_pSession =  pSession;
	return true;
}

AsyncComm::~AsyncComm()
{

}

void AsyncComm::InitNetwork()
{
#if defined( __WIN32__ ) || defined( WIN32 ) || defined( _WIN32 )
	WSADATA wsaData;
	WSAStartup(MAKEWORD(2,0), &wsaData);
#endif
}


bool AsyncComm::Connect( const char * Address )
{
	return true;
}

void AsyncComm::_OnConnect()
{
	// set common parameters on the file descriptor
	_NonblockingFD();

	// Call virtual onconnect
	OnConnect();
}

//disconnect process
void AsyncComm::_Reset()
{
	if (m_fd != INVALID_FD)
		CloseFD();

	m_connected = false;
	m_fd = INVALID_FD;

	if ( !m_SendQueue.empty() )
	{
		DataPacket* packet = NULL;
		while( (packet = m_SendQueue.front()) != NULL )
		{
			m_SendQueue.pop();
			PACKET_DELETE packet;
		}
	}

	m_BytesSent = 0;
    m_BytesRecieved = 0;
	m_lastPong = 0;
	
	m_nRemaining = m_nSize = m_Opcode = 0;
	//TODO:readBuffer clean
	readBuffer.Reset();
	//TODO:writeBuffer clean
	writeBuffer.Reset();
}

bool AsyncComm::Send(const uint8 * Bytes, uint32 Size)
{
	bool rv;

	// This is really just a wrapper for all the burst stuff.
	BurstBegin();
	rv = BurstSend(Bytes, Size);
	if(rv)
		BurstPush();
	BurstEnd();

	return rv;
}

bool AsyncComm::BurstSend(const uint8 * Bytes, uint32 Size)
{
	return writeBuffer.Write(Bytes, Size);
}

//disconnect by client
void AsyncComm::Disconnect()
{
	_Reset();
}

void AsyncComm::WriteCallback()
{
	if(!m_connected)
		return;

	//printf("\nSocket::Writecallback(): sendsize : %u\n", this->m_writeByteCount);
	// We don't want any writes going on while this is happening.
	
	if(writeBuffer.GetContiguiousBytes())
	{
		uint32 flags = 0;
		// attempt to push all the data out in a non-blocking fashion.
		uint32  nLen = (u_long)writeBuffer.GetContiguiousBytes();
		char* pBuff= (char*)writeBuffer.GetBufferStart();


		int nSend = _RawWriteDate( pBuff, nLen );
		if( nSend < 0 )
		{
			OnDisconnect(GetLastError());
			return;
		}
		writeBuffer.Remove(nSend);
		m_BytesSent += nSend;
	}
}

bool AsyncComm::ReadDataCallback()
{
	if(!m_connected)
		return false;

	int flags = 0;
	uint32 nLen	= (u_long)readBuffer.GetSpace();
	char* pBuff = (char*)readBuffer.GetBuffer();	

	int nRecv = _RawReadData( pBuff, nLen) ;
	if( nRecv < 0 )
	{
		OnDisconnect(GetLastError());
		return false;
	}
	else if( nRecv == 0 )
	{
		//_LOG("shutdown...");
		OnDisconnect(0);
		return false;
	}
	GetReadBuffer().IncrementWritten(nRecv);
    m_BytesRecieved += nRecv;


	return true;
}

void AsyncComm::BurstPush()
{
#if !PACK_PACKET
	WriteCallback();
#endif
}

void AsyncComm::Update(int32 deltaTime)
{
	if ( m_fd == INVALID_FD ) 
		return;

	//update queued send packet
	UpdateQueuedPackets();

	int value = 0;
	struct timeval tval = { 0, 0 };
	 
	FD_ZERO( &m_read);
	FD_ZERO( &m_write);
	FD_ZERO( &m_error);
	
	FD_SET( m_fd, &m_read);
	FD_SET( m_fd, &m_write);
	FD_SET( m_fd, &m_error);

	value = select( m_fd + 1, &m_read, &m_write, &m_error, &tval);
	if ( value > 0 )
	{
		if ( !m_connected )
		{
			if ( FD_ISSET( m_fd, &m_read) || FD_ISSET( m_fd, &m_write) || FD_ISSET( m_fd, &m_error))
			{
				OnConnect();
			}
		}
		else
		{
			if (FD_ISSET( m_fd, &m_error) )
			{
				OnDisconnect(0);
				return;
			}

			if ( FD_ISSET( m_fd, &m_read) )
			{
				ReadDataCallback();
				if ( m_connected )
				{
					OnRead();
				}
				else
				{
					return;
				}
			}

			if ( FD_ISSET( m_fd, &m_write) )
			{
				if ( m_connected )
				{
					OnWrite();
				}
				else
				{
					return;
				}
			}
		}
	}
}

// Called when data is received.
void AsyncComm::OnRead() 
{
	for(;;)
	{
		if(m_nRemaining == 0)
		{
			if(GetReadBuffer().GetSize() < sizeof(NetPacketHeader) )
			{
				return;
			}

			NetPacketHeader Header;
			GetReadBuffer().Read((uint8*)&Header, sizeof(NetPacketHeader) );

			Header.size = _BITSWAP16(Header.size);
			Header.cmd	= _BITSWAP16(Header.cmd);
			//GLL_ASSERT(Header.CheckHeader());
			CR_ASSERT(Header.size >= sizeof(NetPacketHeader));
			m_nRemaining = m_nSize = Header.size - sizeof(NetPacketHeader);
			m_Opcode = Header.cmd;
		}

		if(m_nRemaining > 0)
		{
			if( GetReadBuffer().GetSize() < m_nRemaining )
			{
				return;
			}
		}

		DataPacket * Packet = PACKET_NEW DataPacket ( static_cast<uint16>( m_Opcode ));

		if(m_nRemaining > 0)
		{
			GetReadBuffer().Read((uint8*)Packet->GetWriteBuffer(m_nRemaining), m_nRemaining);
		}

		m_nRemaining = m_nSize = m_Opcode = 0;
		switch(Packet->GetOpcode())
		{
		case 0:

		default:
			{
				if(m_pSession) 
					m_pSession->QueuePacket(Packet);
				else 
					PACKET_DELETE Packet;
			}break;
		}
	}
}

void AsyncComm::UpdateQueuedPackets()
{
	if(m_SendQueue.empty())
	{
		return;
	}

	DataPacket * pck;
	while( (pck = m_SendQueue.front()) != NULL )
	{
		/* try to push out as many as you can */
		OUTPACKET_RESULT res = _OutPacket(pck->GetOpcode(), pck->size(), pck->size() ? pck->contents() : NULL);
		switch(res)
		{
		case OUTPACKET_RESULT_SUCCESS:
			{
				PACKET_DELETE pck;
				m_SendQueue.pop();
			}break;

		case OUTPACKET_RESULT_NO_ROOM_IN_BUFFER:
			{

				return;
			}break;

		default:
			{
				/* kill everything in the buffer */
				while( (pck = m_SendQueue.front()) != NULL )
				{
					m_SendQueue.pop();
					PACKET_DELETE pck;
				}
				return;
			}break;
		}
	}
}


void AsyncComm::SendPacket(DataPacket* packet) 
{
	if(!packet) 
		return; 
	OutPacket(packet->GetOpcode(), packet->size(), (packet->size() ? (const void*)packet->contents() : NULL));
}

void AsyncComm::OutPacket(uint16 opcode, size_t len, const void* data)
{
	OUTPACKET_RESULT res;
	if( (len + 10) > CLIENTSOCKET_SENDBUF_SIZE )
	{
		printf("WARNING: Tried to send a packet of %u bytes (which is too large) to a socket. Opcode was: %u (0x%03X)\n", (unsigned int)len, (unsigned int)opcode, (unsigned int)opcode);
		return;
	}

	res = _OutPacket(opcode, len, data);
	if(res == OUTPACKET_RESULT_SUCCESS)
		return;

	if(res == OUTPACKET_RESULT_NO_ROOM_IN_BUFFER)
	{
		/* queue the packet */
		//queueLock.Acquire();
		DataPacket * pck = PACKET_NEW DataPacket(opcode);
		if(len) pck->append((const uint8*)data, len);
		m_SendQueue.push(pck);
		//queueLock.Release();
	}
}

OUTPACKET_RESULT AsyncComm::_OutPacket(uint16 opcode, size_t len, const void* data)
{
	bool rv;
	if(!IsConnected())
		return OUTPACKET_RESULT_NOT_CONNECTED;

	BurstBegin();
	NetPacketHeader Header;
	

	if( GetWriteBuffer().GetSpace() < ( len + sizeof(NetPacketHeader) ) )
	{
		BurstEnd();
		return OUTPACKET_RESULT_NO_ROOM_IN_BUFFER;
	}

	Header.size = _BITSWAP16((uint16)(len + sizeof(NetPacketHeader) ) );
	Header.cmd = _BITSWAP16(opcode);

	// Pass the header to our send buffer
	rv = BurstSend((const uint8*)&Header, sizeof(NetPacketHeader) );

	// Pass the rest of the packet to our send buffer (if there is any)
	if(len > 0 && rv)
	{
		rv = BurstSend((const uint8*)data, (uint32)len);
	}

	if(rv) 
		BurstPush();
	BurstEnd();
	return rv ? OUTPACKET_RESULT_SUCCESS : OUTPACKET_RESULT_SOCKET_ERROR;
}

void AsyncComm::OnWrite() 
{
	BurstBegin();
	if( GetWriteBuffer().GetContiguiousBytes() > 0 )
		WriteCallback();
	BurstEnd();			
}

// Called when a connection is first successfully established.
void AsyncComm::OnConnect() 
{
	int32 errorCode = GetLastError();
	if ( errorCode == 0 )
	{
		m_connected = true;
		_NonblockingFD();
	}
	if ( m_pSession )
		m_pSession->OnConnect( errorCode );
}

//disconnect by server/network,passive disconnect
void AsyncComm::OnDisconnect(int32 errorCode) 
{
	if (errorCode == 0) 
		errorCode = GetLastError();
	_Reset();
	if ( m_pSession )
		m_pSession->OnDisconnect(errorCode);
}